package com.ebupt.migu.fineoperationkconsumermessage.controller;

import com.alibaba.fastjson.JSONObject;
import com.ebupt.migu.common.pojo.Topic;
import com.ebupt.migu.common.push.MessageTopicBean;
import com.ebupt.migu.fineoperationkconsumermessage.service.MessageNoticeService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;
import lombok.extern.slf4j.Slf4j;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.concurrent.TimeUnit;

@Slf4j
@Service
public class KMessageConsumerController {


    @Autowired
    private MessageNoticeService messageNoticeService;


    @Resource
    private RedisTemplate redisTemplate;


    @KafkaListener(topics = Topic.OFFLINE_MESSAGE_TEST, groupId = "RFMENT-OFFLINE-MESSAGE-TEST")
    public void offlineTest(JSONObject bean, Acknowledgment ack) {
        log.info("下发topic --> {}下发策略 -->  {}", Topic.OFFLINE_MESSAGE_TEST, bean.get("strategy"));
        this.consumeMessage(bean, ack);
    }


    @KafkaListener(topics = Topic.OFFLINE_MESSAGE, groupId = "RFMENT-OFFLINE-MESSAGE")
    public void offline(JSONObject bean, Acknowledgment ack) {
        log.info("下发topic --> {},下发策略 --> {}", Topic.OFFLINE_MESSAGE, bean.get("strategy"));
        this.consumeMessage(bean, ack);
    }


    public void consumeMessage(JSONObject bean, Acknowledgment ack) {
        // 上报偏移量
        ack.acknowledge();
        messageNoticeService.doPost(bean.getString("url"), bean.getString("request"));
        if (bean.get("uids") != null) {
            ArrayList arrayList = JSONObject.parseObject(bean.getString("uids"), ArrayList.class);
            for (int i = 0; i < arrayList.size(); i++) {
                messageNoticeService.redis_incr(bean.getString("keyName"));
            }
        } else {
            messageNoticeService.redis_incr(bean.getString("keyName"));
        }
        // 短信下发进度保留3天，下发成功数/失败数保留7天
        redisTemplate.expire(bean.getString("keyName"), 30, TimeUnit.DAYS);
    }


}